package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: xyl
 * @Description: 针对企查查 t_land_mortgage_details 表和中数 t_landmortgage 表进行数据映射开发zs_t_landmortgage
 * @Date: 2021/12/30 16:51
 */
public class TLandMortgageDetails {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_land_mortgage_details").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_landmortgage");
        long zs_count = zs_data.count();
        if(zs_count>0){

            /**
             * t_land_mortgage_details 土地抵押详情
             */
            //获取中数删除数据 --- 中数删除逻辑：公司的任何一条数据发生变化，先删除这个公司相关的该表的所有数据
            Dataset<Row> delete_data = spark.sql("select a.id,a.key_no,a.company_id,a.company_name,a.lm_id,a.land_sign,a.land_no,a.administrative_area,a.acreage,a.address,a.obligee_no,a.usufruct_no, \n" +
                    "a.mortgagor_nature,a.mortgage_purpose,a.nature_and_type,a.mortgage_acreage,a.assessment_price,a.mortgage_price,a.on_board_start_time,a.on_board_end_time, \n" +
                    "a.mortgagor_key_no,a.mortgagor_name,a.mortgage_key_no,a.mortgage_name,a.start_date,a.end_date,  \n" +
                    "c.jobid,'-1' as isadd  \n" +
                    "from a_dc_ep_dwi.t_land_mortgage_details a  \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where company_id is not null and company_id !='') b on a.company_id=b.company_id \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_landmortgage_del where creditcode is not null and creditcode !='')c on upper(trim(b.credit_code)) = upper(trim(c.creditcode)) \n" +
                    "where b.credit_code is not null and c.creditcode is not null  \n");
            delete_data.createOrReplaceTempView("tmp_t_land_mortgage_details");
            spark.sql("insert overwrite table a_dc_ep_incr.t_land_mortgage_details select * from tmp_t_land_mortgage_details");

            //获取中数 数据表中的 旧数据+更新+新增数据
            Dataset<Row> update_data = spark.sql("SELECT " +
                    "       md5(concat(b.company_id,a.RECORD_ID)),\n" +
                    "       b.key_no,\n" +
                    "       b.company_id,\n" +
                    "       b.company_name,\n" +
                    "       \"\",\n" +
                    "       \"\",\n" +
                    "       a.PARCEL_NO,\n" +
                    "       a.DISTRICT,\n" +
                    "       a.LANDAREA,\n" +
                    "       a.PARCEL_LOCATED,\n" +
                    "       a.LAND_OWNER_NO,\n" +
                    "       a.LAND_USE_NO,\n" +
                    "       a.LAND_MORTGAGOR_NATURE,\n" +
                    "       a.MORTGAGE_LAND_USE,\n" +
                    "       a.MORTGAGE_LAND_TYPE,\n" +
                    "       a.MORTGAGE_AREA,\n" +
                    "       a.ASSESS_MONEY,\n" +
                    "       a.MORTGAGE_MONEY,\n" +
                    "       a.START_DATE,\n" +
                    "       a.END_DATE,\n" +
                    "       a.MORTGAGOR_ENTID,\n" +
                    "       a.LAND_MORTGAGOR_NAME,\n" +
                    "       a.MORTGAGEE_ENTID,\n" +
                    "       a.LAND_MORTGAGEE,\n" +
                    "       a.START_DATE,\n" +
                    "       a.END_DATE,\n" +
                    "       a.JOBID,\n" +
                    "       \"0\" as isadd  " +
                    "FROM\n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_landmortgage WHERE MORTGAGOR_CREDITCODE IS NOT NULL AND MORTGAGOR_CREDITCODE !='' and length(trim(MORTGAGOR_CREDITCODE))=18) a\n" +
                    "INNER JOIN\n" +
                    "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') b \n" +
                    "ON upper(trim(a.MORTGAGOR_CREDITCODE)) = upper(trim(b.credit_code)) \n" +
                    "where b.credit_code is not null ").union(spark.sql("SELECT " +
                    "       md5(concat(b.company_id,a.RECORD_ID)),\n" +
                    "       b.key_no,\n" +
                    "       b.company_id,\n" +
                    "       b.company_name,\n" +
                    "       \"\",\n" +
                    "       \"\",\n" +
                    "       a.PARCEL_NO,\n" +
                    "       a.DISTRICT,\n" +
                    "       a.LANDAREA,\n" +
                    "       a.PARCEL_LOCATED,\n" +
                    "       a.LAND_OWNER_NO,\n" +
                    "       a.LAND_USE_NO,\n" +
                    "       a.LAND_MORTGAGOR_NATURE,\n" +
                    "       a.MORTGAGE_LAND_USE,\n" +
                    "       a.MORTGAGE_LAND_TYPE,\n" +
                    "       a.MORTGAGE_AREA,\n" +
                    "       a.ASSESS_MONEY,\n" +
                    "       a.MORTGAGE_MONEY,\n" +
                    "       a.START_DATE,\n" +
                    "       a.END_DATE,\n" +
                    "       a.MORTGAGOR_ENTID,\n" +
                    "       a.LAND_MORTGAGOR_NAME,\n" +
                    "       a.MORTGAGEE_ENTID,\n" +
                    "       a.LAND_MORTGAGEE,\n" +
                    "       a.START_DATE,\n" +
                    "       a.END_DATE,\n" +
                    "       a.JOBID,\n" +
                    "       \"0\" as isadd  " +
                    "FROM\n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_landmortgage WHERE MORTGAGEE_CREDITCODE IS NOT NULL AND MORTGAGEE_CREDITCODE !='' and length(trim(MORTGAGEE_CREDITCODE))=18) a\n" +
                    "INNER JOIN\n" +
                    "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') b \n" +
                    "ON upper(trim(a.MORTGAGEE_CREDITCODE)) = upper(trim(b.credit_code)) \n" +
                    "where b.credit_code is not null "));

            update_data.createOrReplaceTempView("tmp_t_land_mortgage_details1");
            spark.sql("insert into table a_dc_ep_incr.t_land_mortgage_details select * from tmp_t_land_mortgage_details1");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_land_mortgage_details");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
